package com.feidee.fd.sml.algorithm.component.feature

import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.feature.StandardScaler
import org.apache.spark.sql.DataFrame

/**
  * @Author tangjinyuan
  * @Date 2019/03/21 16:33
  * @Description StandardScalerEncoder  特征组件
  * @Reviewer dongguosheng
  */
case class StandardScaleEncoderParam(
                                      override val input_pt: String,
                                      override val output_pt: String,
                                      override val hive_table: String,
                                      override val flow_time: String,
                                      override val inputCol: String,
                                      override val outputCol: String,
                                      override val preserveCols: String,
                                      override val modelPath: String,
                                      // 是否变换为 0 均值，默认 false
                                      withMean: Boolean,
                                      // 是否将数据标准化到单位标准差，默认 true
                                      withStd: Boolean
                                    ) extends FeatureParam {

  def this() = this(null, null, null, null, "input", "features", null, null, false, true)

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("withMean" -> withMean)
    map += ("withStd" -> withStd)
    map
  }
}


class StandardScaleEncoder extends AbstractFeatureEncoder[StandardScaleEncoderParam] {

  override def setUp(param: StandardScaleEncoderParam, data: DataFrame): Array[PipelineStage] = {

    val StandardScaler = new StandardScaler()
      .setWithMean(param.withMean)
      .setWithStd(param.withStd)
      .setInputCol(param.inputCol)
      .setOutputCol(param.outputCol)

    Array(StandardScaler)
  }


}

object StandardScaleEncoder {

  def apply(paramStr: String): Unit = {
    new StandardScaleEncoder()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    StandardScaleEncoder(args(0))
  }
}



